﻿using kratos;
using RingBuffer;
using System;
using System.Net;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Runtime.CompilerServices;
using System.Threading.Tasks;

namespace rpc {

using GlobalProxyCallManager = Singleton<ManagerThreaded<uint, ProxyCall>>;
using GlobalRpc = Singleton<Rpc>;

/// <summary>
/// 默认日志实现
/// </summary>
internal class NullLogger : ILogger {
  public void Debug(string log) {}

  public void Fail(string log) {}

  public void Fatal(string log) {}

  public void Info(string log) {}

  public void Warn(string log) {}
}

/// <summary>
/// 日志等级
/// </summary>
internal enum LogLevel {
  debug = 1,
  info,
  warn,
  fail,
  fatal,
}

/// <summary>
/// RPC对象类型，对象在逻辑线程和网络线程间传递
/// </summary>
enum RpcObjectType {
  none = 0,
  /// <summary>
  /// RPC调用
  /// </summary>
  rpc_call = 1,
  /// <summary>
  /// RPC调用返回
  /// </summary>
  rpc_return,
}

/// <summary>
/// RPC错误码
/// </summary>
enum RpcError {
  /// <summary>
  /// 调用成功
  /// </summary>
  rpc_ok = 1,
  /// <summary>
  /// RPC未找到对应服务或方法
  /// </summary>
  rpc_not_found = 2,
  /// <summary>
  /// 调用对端服务方法时发生异常
  /// </summary>
  rpc_exception = 3,
  /// <summary>
  /// 调用超时
  /// </summary>
  rpc_timeout = 4,
}

/// <summary>
/// RPC远程调用参数对象，用来包裹一次RPC调用所有参数
/// </summary>
public interface CallParam {
  /// <summary>
  /// 单次调用ID, 对于客户端来说单进程唯一即可
  /// </summary>
  uint CallID {
    get;
    set;
  }
  /// <summary>
  /// 服务UUID
  /// </summary>
  ulong ServiceUUID { get; }
  /// <summary>
  /// 方法ID
  /// </summary>
  uint MethodID { get; }
  /// <summary>
  /// 服务实例ID
  /// </summary>
  uint ServiceID {
    get;
    set;
  }
  /// <summary>
  /// 所有参数序列化后的数据长度，字节
  /// </summary>
  uint Length { get; }
  /// <summary>
  /// 将网络缓冲区内的数据反序列化
  /// </summary>
  /// <param name="rb">网络缓冲区</param>
  /// <param name="length">需要反序列化的数据长度</param>
  void deserialize(GrowingRingBuffer<byte> rb, uint length);
  /// <summary>
  /// 将对象序列化为字节流
  /// </summary>
  /// <returns>字节流</returns>
  byte[] serialize();
  /// <summary>
  /// 是否调用本地服务进行测试
  /// </summary>
  bool IsDebugLoop {
    get;
    set;
  }
  /// <summary>
  /// 回收方法
  /// </summary>
  RecycleFunc RecycleMethod { get; }
}

/// <summary>
/// 一次RPC调用的返回值包裹对象
/// </summary>
public interface CallReturn {
  /// <summary>
  /// 单次调用ID, 对于客户端来说单进程唯一即可
  /// </summary>
  uint CallID {
    get;
    set;
  }
  /// <summary>
  /// 服务实例ID
  /// </summary>
  uint ServiceID {
    get;
    set;
  }
  /// <summary>
  /// 返回值序列化后的数据长度，字节
  /// </summary>
  uint Length { get; }
  /// <summary>
  /// 本次调用的错误ID
  /// </summary>
  uint ErrorID {
    get;
    set;
  }
  /// <summary>
  /// 将网络缓冲区内的数据反序列化
  /// </summary>
  /// <param name="rb">网络缓冲区</param>
  /// <param name="length">需要反序列化的数据长度</param>
  void deserialize(GrowingRingBuffer<byte> rb, uint length);
  /// <summary>
  /// 将对象序列化为字节流
  /// </summary>
  /// <returns>字节流</returns>
  byte[] serialize();
  /// <summary>
  /// 回收方法
  /// </summary>
  RecycleFunc RecycleMethod { get; }
}

/// <summary>
/// 单例模板类
/// </summary>
/// <typeparam name="T"></typeparam>
internal class Singleton<T> where T : new () {
  static T real_singleton_;
  public static T Instance {
    get {
      if (real_singleton_ == null) {
        real_singleton_ = new T();
      }
      return real_singleton_;
    }
  }
}

/// <summary>
/// 通用单线程对象管理器
/// </summary>
/// <typeparam name="KeyT">管理器key</typeparam>
/// <typeparam name="ValueT">管理器value</typeparam>
internal class Manager<KeyT, ValueT> {
  Dictionary<KeyT, ValueT> object_dict_ = new Dictionary<KeyT, ValueT>();
  public ValueT get(KeyT key) {
    ValueT value;
    if (object_dict_.TryGetValue(key, out value)) {
      return value;
    }
    return default(ValueT);
  }

  public bool Add(KeyT key, ValueT value) {
    if (value == null) {
      return false;
    }
    ValueT exist_value;
    if (object_dict_.TryGetValue(key, out exist_value)) {
      return false;
    }
    object_dict_.Add(key, value);
    return true;
  }

  public bool Remove(KeyT key) { return object_dict_.Remove(key); }

  public Dictionary<KeyT, ValueT> Dict => object_dict_;
}

/// <summary>
/// 通用多线程对象管理器
/// </summary>
/// <typeparam name="KeyT">管理器key</typeparam>
/// <typeparam name="ValueT">管理器value</typeparam>
internal class ManagerThreaded<KeyT, ValueT> {
  Dictionary<KeyT, ValueT> object_dict_ = new Dictionary<KeyT, ValueT>();
  object lock_obj_ = new object();
  public ValueT Get(KeyT key) {
    ValueT value;
    lock(lock_obj_) {
      if (object_dict_.TryGetValue(key, out value)) {
        return value;
      }
    }
    return default(ValueT);
  }

  public bool Add(KeyT key, ValueT value) {
    if (value == null) {
      return false;
    }
    ValueT exist_value;
    lock(lock_obj_) {
      if (object_dict_.TryGetValue(key, out exist_value)) {
        return false;
      }
      object_dict_.Add(key, value);
    }
    return true;
  }

  public bool Remove(KeyT key) {
    lock(lock_obj_) { return object_dict_.Remove(key); }
  }

  public void Clear() {
    lock(lock_obj_) { object_dict_.Clear(); }
  }
}

public delegate CallParam NewParamByUUID();

/// <summary>
/// 调用ID生成器，在主线程内产生
/// </summary>
internal static class CallIDGenerator {
  static uint call_id_ = 1;
  public static uint generate() { return call_id_++; }
}

public delegate CallReturn NewReturnByUUID();

/// <summary>
/// 客户端使用的单次RPC调用对象
/// </summary>
public interface ProxyCall {
  /// <summary>
  /// 服务UUID
  /// </summary>
  ulong ServiceUUID { get; }
  /// <summary>
  /// 调用ID
  /// </summary>
  uint CallID { get; }
  /// <summary>
  /// 方法ID
  /// </summary>
  uint MethodID { get; }
  /// <summary>
  /// 协程Continuation
  /// </summary>
  Action Action { get; }
  /// <summary>
  /// 错误ID
  /// </summary>
  uint ErrorID {
    get;
    set;
  }
  /// <summary>
  /// 是否超时
  /// </summary>
  bool IsTimeout { get; }
  /// <summary>
  /// 超时检查
  /// </summary>
  /// <param name="now">当前时间戳，毫秒</param>
  void CheckExpired(long now);
}

/// <summary>
/// 本地服务管理器
/// </summary>
internal class ServiceManager {
  /// <summary>
  /// {服务实例ID，服务实例}
  /// </summary>
  Manager<uint, IService> service_instance_manager_ =
      new Manager<uint, IService>();
  /// <summary>
  /// {服务UUID，服务实例}
  /// </summary>
  Manager<ulong, IService> service_uuid_manager_ =
      new Manager<ulong, IService>();
  /// <summary>
  /// 注册服务实例
  /// TODO 多实例
  /// </summary>
  /// <param name="uuid">服务UUID</param>
  /// <param name="service">服务实例</param>
  /// <returns></returns>
  public bool RegisterService(ulong uuid, IService service) {
    if (service_uuid_manager_.Add(uuid, service)) {
      return service_instance_manager_.Add(service.ID, service);
    }
    return false;
  }
  /// <summary>
  /// 通过UUID获取服务实例
  /// </summary>
  /// <param name="uuid">服务UUID</param>
  /// <returns></returns>
  public IService GetServiceByUUID(ulong uuid) {
    return service_uuid_manager_.get(uuid);
  }
  /// <summary>
  /// 通过服务实例ID获取服务实例
  /// </summary>
  /// <param name="id">服务实例ID</param>
  /// <returns></returns>
  public IService GetServiceByID(uint id) {
    return service_instance_manager_.get(id);
  }
  /// <summary>
  /// 调用所有服务的OnTick方法
  /// </summary>
  /// <param name="now"></param>
  public void Update(long now) {
    foreach (var inst in service_instance_manager_.Dict) {
      inst.Value.OnTick(now);
    }
  }
}
/// <summary>
/// 对象序列化/反序列化器
/// </summary>
internal class RpcSerializer : Serializer {
  /// <summary>
  /// RPC协议头长度
  /// </summary>
  const int HEADER_LENGTH = 8;
  /// <summary>
  /// RPC调用返回协议头长度
  /// </summary>
  const int RPC_RETURN_HEADER_LENGTH = 12;
  /// <summary>
  /// RPC调用协议头长度
  /// </summary>
  const int RPC_CALL_HEADER_LENGTH = 20;
  byte[] header_data_ = new byte[HEADER_LENGTH];
  byte[] rpc_call_header_ = new byte[RPC_CALL_HEADER_LENGTH];
  byte[] rpc_return_header_ = new byte[RPC_RETURN_HEADER_LENGTH];
  /// <summary>
  /// RPC协议缓冲区，默认1M，自增长
  /// </summary>
  GrowingRingBuffer<byte> recv_buffer_ =
      new GrowingRingBuffer<byte>(1024 * 1024);
  /// <summary>
  /// 检测协议流，拆包
  /// </summary>
  /// <param name="rb">协议流</param>
  /// <param name="length">协议体长度，字节</param>
  /// <param name="type">对象类型</param>
  /// <returns></returns>
  private bool CheckStream(GrowingRingBuffer<byte> rb, out uint length,
                           out RpcObjectType type) {
    if (rb.Count < HEADER_LENGTH) {
      length = 0;
      type = 0;
      return false;
    }
    // 拷贝RPC包头
    rb.Peek(header_data_, HEADER_LENGTH);
    // 包长度
    length = (uint) IPAddress.NetworkToHostOrder(
        BitConverter.ToInt32(header_data_, 0));
    // 类型
    type = (RpcObjectType) IPAddress.NetworkToHostOrder(
        BitConverter.ToInt32(header_data_, 4));
    if (rb.Count < length) {
      // 包不完整
      length = 0;
      type = 0;
      return false;
    }
    length -= HEADER_LENGTH;
    OnEat(HEADER_LENGTH);
    // 已经接收到了一个完整包
    return true;
  }
  public override object deserialize(byte[] data, int size) {
    for (int i = 0; i < size; i++) {
      recv_buffer_.Put(data[i]);
    }
    uint length;
    RpcObjectType type;
    Object obj = null;
    uint before_enter_size = 0;
    uint after_leave_size = 0;
    if (CheckStream(recv_buffer_, out length, out type)) {
      obj = null;
      before_enter_size = (uint) recv_buffer_.Count;
      switch (type) {
      case RpcObjectType.rpc_call:
        obj = OnRpcCall(length);
        break;
      case RpcObjectType.rpc_return:
        obj = OnRpcReturn(length);
        break;
      default:
        break;
      }
      after_leave_size = (uint) recv_buffer_.Count;
      uint gap = before_enter_size - after_leave_size;
      if (length != gap) {
        OnEat(length - gap);
      }
    }
    return obj;
  }
  /// <summary>
  /// 从输入字节流内获取调用参数并调用本地服务
  /// </summary>
  /// <param name="length">需要解析的字节流长度，字节</param>
  /// <returns>调用参数对象</returns>
  private Object OnRpcCall(uint length) {
    // 解析包头
    recv_buffer_.Peek(rpc_call_header_, RPC_CALL_HEADER_LENGTH);
    ulong service_uuid = (ulong) IPAddress.NetworkToHostOrder(
        BitConverter.ToInt64(rpc_call_header_, 0));
    uint service_id = (uint) IPAddress.NetworkToHostOrder(
        BitConverter.ToInt32(rpc_call_header_, 8));
    uint call_id = (uint) IPAddress.NetworkToHostOrder(
        BitConverter.ToInt32(rpc_call_header_, 12));
    uint method_id = (uint) IPAddress.NetworkToHostOrder(
        BitConverter.ToInt32(rpc_call_header_, 16));
    OnEat(RPC_CALL_HEADER_LENGTH);
    return OnStubCall(service_uuid, service_id, call_id, method_id,
                      length - RPC_CALL_HEADER_LENGTH);
  }
  /// <summary>
  /// 从字节流内建立调用参数对象
  /// </summary>
  /// <param name="service_uuid">服务UUID</param>
  /// <param name="service_id">服务实例ID</param>
  /// <param name="call_id">调用ID</param>
  /// <param name="method_id">方法ID</param>
  /// <param name="object_length">参数对象字节流长度，字节</param>
  /// <returns></returns>
  private Object OnStubCall(ulong service_uuid, uint service_id, uint call_id,
                            uint method_id, uint object_length) {
    // 建立协议包裹对象，包裹对象应该是代码生成的，里面包含了参数对象，用于stub调用
    CallParam param =
        Singleton<ParamRegister>.Instance.Create(service_uuid, method_id);
    if (param == null) {
      GlobalRpc.Instance.WriteLogToMain(
          LogLevel.warn,
          $"[RPC]被调用服务未找到，服务UUID:{service_uuid}, 方法ID:{method_id}");
      return null;
    }
    param.ServiceID = service_id;
    param.CallID = call_id;
    // 从字节流内获取调用参数
    param.deserialize(recv_buffer_, object_length);
    return param;
  }
  /// <summary>
  /// 从字节流内解析返回值对象
  /// </summary>
  /// <param name="length">需要解析的长度，字节</param>
  /// <returns>返回值对象</returns>
  private Object OnRpcReturn(uint length) {
    // 解析包头
    recv_buffer_.Peek(rpc_return_header_, RPC_RETURN_HEADER_LENGTH);
    uint service_id = (uint) IPAddress.NetworkToHostOrder(
        BitConverter.ToInt32(rpc_return_header_, 0));
    uint call_id = (uint) IPAddress.NetworkToHostOrder(
        BitConverter.ToInt32(rpc_return_header_, 4));
    uint error_id = (uint) IPAddress.NetworkToHostOrder(
        BitConverter.ToInt32(rpc_return_header_, 8));
    OnEat(RPC_RETURN_HEADER_LENGTH);
    return OnStubCallReturn(service_id, call_id, error_id,
                            length - RPC_RETURN_HEADER_LENGTH);
  }

  /// <summary>
  /// 从字节流内建立返回值对象
  /// </summary>
  /// <param name="service_id">服务实例ID</param>
  /// <param name="call_id">调用ID</param>
  /// <param name="error_id">错误ID</param>
  /// <param name="object_length">需要解析的长度，字节</param>
  /// <returns>返回值对象</returns>
  private Object OnStubCallReturn(uint service_id, uint call_id, uint error_id,
                                  uint object_length) {
    // 建立协议包裹对象，包裹对象应该是代码生成的，里面包含了返回值对象，用于proxy
    // call返回值调用
    ProxyCall call = GlobalProxyCallManager.Instance.Get(call_id);
    if (call == null) {
      GlobalRpc.Instance.WriteLogToMain(LogLevel.debug,
                                        "[RPC]RPC CALL返回，但ProxyCall未找到");
      return null;
    }
    CallReturn call_return = Singleton<ReturnRegister>.Instance.Create(
        call.ServiceUUID, call.MethodID);
    if (call_return == null) {
      GlobalRpc.Instance.WriteLogToMain(
          LogLevel.warn,
          $"[RPC]RPC CallReturn未找到, 服务UUID:{call.ServiceUUID}, 服务方法ID:{call.MethodID}");
      return null;
    }
    call_return.ServiceID = service_id;
    call_return.CallID = call_id;
    call_return.ErrorID = error_id;
    // 从字节流获取返回值
    call_return.deserialize(recv_buffer_, object_length);
    return call_return;
  }

  private void OnEat(uint length) {
    for (uint i = 0; i < length; i++) {
      recv_buffer_.Get();
    }
  }

  private byte[] OnSerialize(Object obj) {
    if (obj is CallParam) {
      return OnProxyCall((CallParam) obj);
    } else if (obj is CallReturn) {
      return OnProxyReturn((CallReturn) obj);
    } else {
      return null;
    }
  }
  /// <summary>
  /// 将调用参数对象序列化为字节流
  /// </summary>
  /// <param name="param">参数对象</param>
  /// <returns>字节流</returns>
  private byte[] OnProxyCall(CallParam param) {
    // --------------------
    // RPC hearder         |
    // --------------------
    // RPC call header     |
    // --------------------
    // RPC parameter body  |
    // --------------------

    // 构造包体，获取参数包体长度
    byte[] body = param.serialize();
    byte[] packet_bytes =
        new byte[param.Length + HEADER_LENGTH + RPC_CALL_HEADER_LENGTH];
    // 拷贝参数包体
    if (body != null) {
      Array.Copy(body, 0, packet_bytes, HEADER_LENGTH + RPC_CALL_HEADER_LENGTH,
                 param.Length);
      GlobalBytePool.Pool.Return(body);
    }
    var header_length = IPAddress.HostToNetworkOrder(
        (int)(param.Length + HEADER_LENGTH + RPC_CALL_HEADER_LENGTH));
    var type = IPAddress.HostToNetworkOrder((int) RpcObjectType.rpc_call);
    byte[] length_bytes = BitConverter.GetBytes(header_length);
    byte[] type_bytes = BitConverter.GetBytes(type);
    // 拷贝RPC包头
    length_bytes.CopyTo(packet_bytes, 0);
    type_bytes.CopyTo(packet_bytes, 4);
    byte[] uuid_bytes = BitConverter.GetBytes(
        IPAddress.HostToNetworkOrder((long) param.ServiceUUID));
    byte[] service_id_bytes = BitConverter.GetBytes(
        IPAddress.HostToNetworkOrder((int) param.ServiceID));
    byte[] call_id_bytes =
        BitConverter.GetBytes(IPAddress.HostToNetworkOrder((int) param.CallID));
    byte[] method_id_bytes = BitConverter.GetBytes(
        IPAddress.HostToNetworkOrder((int) param.MethodID));
    // 回收对象
    param.RecycleMethod(param);
    // 构造RPC调用包头
    uuid_bytes.CopyTo(packet_bytes, 8);
    service_id_bytes.CopyTo(packet_bytes, 16);
    call_id_bytes.CopyTo(packet_bytes, 20);
    method_id_bytes.CopyTo(packet_bytes, 24);
    return packet_bytes;
  }
  /// <summary>
  /// 将调用结果对象序列化为字节流
  /// </summary>
  /// <param name="call_return">结果对象</param>
  /// <returns>字节流</returns>
  private byte[] OnProxyReturn(CallReturn call_return) {
    // ----------------------
    // RPC hearder           |
    // ----------------------
    // RPC call return header|
    // ----------------------
    // RPC return body       |
    // ----------------------

    // 构造包体
    byte[] body = call_return.serialize();
    byte[] packet_bytes =
        new byte[call_return.Length + HEADER_LENGTH + RPC_RETURN_HEADER_LENGTH];
    // 写入Rpc return body
    if (body != null) {
      Array.Copy(body, 0, packet_bytes,
                 HEADER_LENGTH + RPC_RETURN_HEADER_LENGTH, call_return.Length);
      GlobalBytePool.Pool.Return(body);
    }
    byte[] length_bytes =
        BitConverter.GetBytes((uint) IPAddress.HostToNetworkOrder(
            (int) call_return.Length + HEADER_LENGTH +
            RPC_RETURN_HEADER_LENGTH));
    byte[] type_bytes = BitConverter.GetBytes(
        (uint) IPAddress.HostToNetworkOrder((int) RpcObjectType.rpc_return));
    // 写入RPC包头
    length_bytes.CopyTo(packet_bytes, 0);
    type_bytes.CopyTo(packet_bytes, 4);
    // 构造包头
    byte[] service_id_bytes = BitConverter.GetBytes(
        (uint) IPAddress.HostToNetworkOrder((int) call_return.ServiceID));
    byte[] call_id_bytes = BitConverter.GetBytes(
        (uint) IPAddress.HostToNetworkOrder((int) call_return.CallID));
    byte[] error_id_bytes = BitConverter.GetBytes(
        (uint) IPAddress.HostToNetworkOrder((int) call_return.ErrorID));
    // 回收对象
    call_return.RecycleMethod(call_return);
    // 写入Call return body
    service_id_bytes.CopyTo(packet_bytes, 8);
    call_id_bytes.CopyTo(packet_bytes, 12);
    error_id_bytes.CopyTo(packet_bytes, 16);
    return packet_bytes;
  }

  public override byte[] serialize(object obj) {
    // 对象序列化为字节流
    return OnSerialize(obj);
  }
}

/// <summary>
/// 网络事件处理器
/// </summary>
internal class NetHander : IClientHandler {
  Rpc rpc_ = null;

  public NetHander(Rpc rpc) { rpc_ = rpc; }

  public override void OnClose(IClient client) {}

  public override void OnConnect(IClient client) {
    // 连接成功
    rpc_.ConnetingAwaiter.finished();
  }

  public override void OnConnectFail(IClient client) {
    // 连接失败
    rpc_.ConnetingAwaiter.finished();
  }

  public override void OnReceive(IClient client, GrowingRingBuffer<byte> rb) {
    // 不处理字节流，只处理RPC对象
    rb.Clear();
  }

  public override void OnReceiveObject(IClient client, object obj) {
    // 接收到RPC对象
    rpc_.OnMessage(obj);
  }

  public override void OnReceiveTimeout(IClient client) {}

  public override void OnReconnect(IClient client) {}

  public override void OnSendFail(IClient client) {}
}

internal delegate void CallServiceMethod(IClient client, CallParam param,
                                         IService service);

/// <summary>
/// 服务调用方法管理器
/// </summary>
internal class ServiceCaller {
  /// <summary>
  /// 本地服务表, { 服务UUID， {服务实例ID, 服务实例调用方法} }
  /// </summary>
  Dictionary<ulong, Dictionary<uint, CallServiceMethod>> method_callers_ =
      new Dictionary<ulong, Dictionary<uint, CallServiceMethod>>();
  /// <summary>
  /// 调用本地服务
  /// </summary>
  /// <param name="client">客户端实例</param>
  /// <param name="param">参数对象</param>
  /// <param name="service">被调用的服务</param>
  internal void Call(IClient client, CallParam param, IService service) {
    Dictionary<uint, CallServiceMethod> methods = null;
    if (!method_callers_.TryGetValue(param.ServiceUUID, out methods)) {
      return;
    }
    CallServiceMethod method = null;
    if (!methods.TryGetValue(param.MethodID, out method)) {
      return;
    }
    try {
      method(client, param, service);
    } catch (Exception e) {
      GlobalRpc.Instance.Logger.Warn(
          $"[RPC]调用服务方法抛出异常:{e.Message}, 服务UUID:{service.UUID}, 服务方法ID:{param.MethodID}");
    } finally {
      param.RecycleMethod(param);
    }
  }
  /// <summary>
  /// 添加服务调用方法
  /// </summary>
  /// <param name="uuid">服务UUID</param>
  /// <param name="method_id">方法ID</param>
  /// <param name="method">调用方法</param>
  internal void Add(ulong uuid, uint method_id, CallServiceMethod method) {
    Dictionary<uint, CallServiceMethod> methods = null;
    if (!method_callers_.TryGetValue(uuid, out methods)) {
      method_callers_.Add(uuid, new Dictionary<uint, CallServiceMethod>());
      method_callers_ [uuid]
          .Add(method_id, method);
    } else {
      methods.Add(method_id, method);
    }
  }
}

/// <summary>
/// IRpc接口实现类
/// </summary>
public class Rpc : IRpc {
  /// <summary>
  /// 日志
  /// </summary>
  ILogger logger_ = new NullLogger();
  /// <summary>
  /// 网络客户端
  /// </summary>
  IClient client_ = null;
  /// <summary>
  /// 连接协程类
  /// </summary>
  ConnectAwaiter connect_awaiter_;
  /// <summary>
  /// 已经完成的调用，临时表
  /// </summary>
  List<ProxyCall> timedout_calls = new List<ProxyCall>();
  /// <summary>
  /// 未完成的调用
  /// </summary>
  Dictionary<uint, ProxyCall> waiting_calls = new Dictionary<uint, ProxyCall>();
  /// <summary>
  /// 日志信息
  /// </summary>
  internal struct Log {
    public LogLevel level;
    public string line;
  }
  /// <summary>
  /// 网络线程到主线程的日志队列
  /// </summary>
  ConcurrentQueue<Log> log_queue_ = new ConcurrentQueue<Log>();
  /// <summary>
  /// 对端地址
  /// </summary>
  string host_;
  /// <summary>
  /// 对端端口
  /// </summary>
  int port_ = 0;
  /// <summary>
  /// 连接超时，毫秒
  /// </summary>
  long connect_timeout_ = 5000;
  /// <summary>
  /// 是否自动重连
  /// </summary>
  bool auto_reconnect_ = true;
  /// <summary>
  /// 自动重连间隔，毫秒
  /// </summary>
  long auto_reconnect_intval_ = 5000;
  /// <summary>
  /// 上次自动重连发起的时间戳
  /// </summary>
  long last_auto_reconnect_ms = 0;
  /// <summary>
  /// 上次运行Update时的时间戳，毫秒
  /// </summary>
  long last_tick_ = 0;
  /// <summary>
  /// 使用使用Tick代替时间戳
  /// </summary>
  bool use_tick_as_timestamp_ = false;
  /// <summary>
  /// 构造
  /// </summary>
  public Rpc() {
    // 注册RPC
    try {
      RpcRegister.DoRegister();
    } catch (Exception e) {
      GlobalRpc.Instance.Logger.Fail($"[RPC]注册基础设施抛出异常:{e.Message}");
    }
  }
  /// <summary>
  /// 子线程写日志
  /// </summary>
  /// <param name="level">日志等级</param>
  /// <param name="line">日志行</param>
  internal void WriteLogToMain(LogLevel level, string line) {
    var log = new Log();
    log.level = level;
    log.line = line;
    log_queue_.Enqueue(log);
  }
  /// <summary>
  /// 添加未完成调用
  /// </summary>
  /// <param name="call"></param>
  internal void AddWaiting(ProxyCall call) {
    waiting_calls.Add(call.CallID, call);
  }
  /// <summary>
  /// 删除未完成调用
  /// </summary>
  /// <param name="call"></param>
  internal bool RemoveWaiting(ProxyCall call) {
    return waiting_calls.Remove(call.CallID);
  }
  /// <summary>
  /// RPC协议处理
  /// </summary>
  /// <param name="obj"></param>
  internal void OnMessage(Object obj) {
    // 逻辑线程内调用
    if (obj is CallParam) {
      CallParam param = (CallParam) obj;
      // 调用本地服务
      IService service = null;
      if (param.ServiceID != 0) {
        // 调用指定ID的服务
        service =
            Singleton<ServiceManager>.Instance.GetServiceByID(param.ServiceID);

      } else {
        // 调用指定UUID的服务
        service = Singleton<ServiceManager>.Instance.GetServiceByUUID(
            param.ServiceUUID);
      }
      if (service == null) {
        GlobalRpc.Instance.Logger.Warn(
            $"[RPC]被调用服务未找到,服务UUID:{param.ServiceUUID},服务实例ID:{param.ServiceID}");
        return;
      }
      Singleton<ServiceCaller>.Instance.Call(client_, param, service);
    } else if (obj is CallReturn) {
      // 服务调用返回
      CallReturn call_return = (CallReturn) obj;
      ProxyCall call = GlobalProxyCallManager.Instance.Get(call_return.CallID);
      try {
        if (call == null) {
          GlobalRpc.Instance.Logger.Warn(
              $"[RPC]被调用服务返回，但未找到ProxyCall,调用ID:{call_return.CallID},服务实例ID:{call_return.ServiceID}");
          return;
        }
        call.ErrorID = call_return.ErrorID;
        // 恢复执行
        if (call.Action != null) {
          call.Action.Invoke();
        }
        Singleton<Rpc>.Instance.RemoveWaiting(call);
        GlobalProxyCallManager.Instance.Remove(call_return.CallID);
      } finally {
        // 回收对象
        call_return.RecycleMethod(call_return);
      }
    }
  }

  public bool Initialize(ILogger logger) {
    logger_ = logger;
    return true;
  }

  internal ILogger Logger => logger_;

  internal IClient Client => client_;
  public long Tick { get => last_tick_; }

  public bool UseTickAsTimestamp {
    get => use_tick_as_timestamp_;
    set => use_tick_as_timestamp_ = value;
  }

  public bool AutoReconnect {
    get { return auto_reconnect_; }
    set { auto_reconnect_ = value; }
  }
  public long AutoReconnectIntvalMS {
    get { return auto_reconnect_intval_; }
    set {
      if (value == 0) {
        auto_reconnect_intval_ = 5000;
      } else {
        auto_reconnect_intval_ = value;
      }
    }
  }

  public bool IsConnected {
    get {
      if (client_ == null) {
        return false;
      }
      return client_.Connected;
    }
  }
  /// <summary>
  /// 用于连接的协程等待器
  /// </summary>
  internal ConnectAwaiter ConnetingAwaiter {
    get { return connect_awaiter_; }
    set { connect_awaiter_ = value; }
  }

  public async Task<bool> Connect(string host, int port, long timeout) {
    if (client_ != null && client_.Connected) {
      return true;
    }
    if (host == null) {
      return false;
    }
    host_ = host;
    port_ = port;
    connect_timeout_ = timeout;
    client_ = ClientFactory.Create();
    // TODO 加解密，等待C++完成开发
    var retval = client_.Connect(host, port, new NetHander(this));
    if (!retval) {
      return false;
    }
    client_.ObjectSerializer = new RpcSerializer();
    connect_awaiter_ = new ConnectAwaiter();
    return await connect_awaiter_;
  }

  public async Task<bool> Reconnect() {
    if (client_ != null) {
      if (client_.Connected) {
        return true;
      }
    }
    if (auto_reconnect_) {
      return false;
    }
    return await Connect(host_, port_, connect_timeout_);
  }

  public void Close() {
    if (client_ != null) {
      client_.Close();
      client_ = null;
    }
    timedout_calls.Clear();
    waiting_calls.Clear();
    GlobalProxyCallManager.Instance.Clear();
  }

  /// <summary>
  /// 处理所有超时的调用
  /// </summary>
  /// <param name="now">当前时间戳，毫秒</param>
  private void DealWithPendingCalls(long now) {
    // 检查所有proxy call是否超时
    timedout_calls.Clear();
    foreach (var item in waiting_calls) {
      item.Value.CheckExpired(now);
      if (item.Value.IsTimeout) {
        logger_.Warn(
            $"[RPC]调用超时，服务UUID:{item.Value.ServiceUUID},方法ID:{item.Value.MethodID}");
        timedout_calls.Add(item.Value);
        GlobalProxyCallManager.Instance.Remove(item.Key);
      }
    }
    for (int i = 0; i < timedout_calls.Count; i++) {
      // 超时，唤醒协程
      var action = timedout_calls[i].Action;
      action.Invoke();
    }
  }
  /// <summary>
  /// 处理自动重连
  /// </summary>
  /// <param name="now">当前时间戳，毫秒</param>
  private async void DealWithAutoReconnect(long now) {
    if (client_ == null) {
      return;
    }
    if (client_.Connected) {
      return;
    }
    if (auto_reconnect_intval_ < connect_timeout_) {
      connect_timeout_ = auto_reconnect_intval_;
    }
    if (last_auto_reconnect_ms == 0) {
      last_auto_reconnect_ms = now;
    }
    if (now - last_auto_reconnect_ms < auto_reconnect_intval_) {
      return;
    }
    last_auto_reconnect_ms = now;
    await Reconnect();
  }
  /// <summary>
  /// 写入工作线程发送来的日志
  /// </summary>
  private void DealWithLog() {
    if (logger_ == null) {
      return;
    }
    Log log;
    while (log_queue_.TryDequeue(out log)) {
      switch (log.level) {
      case LogLevel.debug:
        logger_.Debug(log.line);
        break;
      case LogLevel.info:
        logger_.Info(log.line);
        break;
      case LogLevel.warn:
        logger_.Warn(log.line);
        break;
      case LogLevel.fail:
        logger_.Fail(log.line);
        break;
      case LogLevel.fatal:
        logger_.Fatal(log.line);
        break;
      default:
        break;
      }
    }
  }

  public void Update() {
    // TODO 重试次数
    long now = Util.GetNowSystemMS();
    last_tick_ = now;
    DealWithPendingCalls(now);
    if (client_ != null) {
      client_.Update();
    }
    // 调用本地服务主循环
    Singleton<ServiceManager>.Instance.Update(now);
    // 处理自动重连
    if (auto_reconnect_) {
      DealWithAutoReconnect(now);
    }
    // 日志
    DealWithLog();
  }
}

public static class Util {
  /// <summary>
  /// 获取当前时间戳，毫秒
  /// </summary>
  /// <returns></returns>
  public static long GetNowMS() {
    if (GlobalRpc.Instance.UseTickAsTimestamp) {
      return GlobalRpc.Instance.Tick;
    }
    return GetNowSystemMS();
  }

  public static long GetNowSystemMS() {
    return (DateTime.Now.ToUniversalTime().Ticks - 621355968000000000) / 10000;
  }
}

/// <summary>
/// 连接远程主机等待类
/// </summary>
internal class ConnectAwaiter : INotifyCompletion {
  /// <summary>
  /// 协程调度类
  /// </summary>
  Action action_;
  public ConnectAwaiter() {
    action_ = null;
    GlobalRpc.Instance.ConnetingAwaiter = this;
  }
  public ConnectAwaiter GetAwaiter() { return this; }
  public void OnCompleted(Action continuation) { action_ = continuation; }
  public bool IsCompleted => false;
  public bool GetResult() { return GlobalRpc.Instance.Client.Connected; }
  public void finished() {
    if (action_ != null) {
      action_.Invoke();
      action_ = null;
    }
  }
}

/// <summary>
/// RPC调用等待类
/// </summary>
internal class ProxyCallAwaiter : ProxyCall, INotifyCompletion, IRecycle {
  // <summary>
  /// 调用超时
  /// </summary>
  bool timeout_ = false;
  /// <summary>
  /// 服务UUID
  /// </summary>
  ulong service_uuid_ = 0;
  /// <summary>
  /// 调用的方法序号
  /// </summary>
  uint method_id_ = 0;
  /// <summary>
  /// RPC调用ID
  /// </summary>
  uint call_id_ = 0;
  /// <summary>
  /// 本次调用对端返回的错误码
  /// </summary>
  uint error_id_ = (uint) RpcError.rpc_ok;
  /// <summary>
  /// 调用超时时间戳，毫秒
  /// </summary>
  long dead_line_ = 0;
  /// <summary>
  /// 调度器对象
  /// </summary>
  Action action_ = null;
  /// <summary>
  /// 是否是本地测试
  /// </summary>
  bool debug_loop_ = false;
  public ProxyCallAwaiter() {}
  public ProxyCallAwaiter GetAwaiter() { return this; }
  public void OnCompleted(Action continuation) {
    action_ = continuation;
    // 保存RPC调用到管理器
    GlobalProxyCallManager.Instance.Add(CallID, this);
    if (!debug_loop_) {
      // 保存到超时管理
      Singleton<Rpc>.Instance.AddWaiting(this);
    }
  }
  public bool IsCompleted => debug_loop_;
  // 检测是否超时
  public void CheckExpired(long now) { timeout_ = (dead_line_ < now); }
  /// <summary>
  /// 没有返回值
  /// </summary>
  public void GetResult() { /*ObjectPool<ProxyCallAwaiter>.Recycle(this);*/
  }
  public void OnReset() {
    timeout_ = false;
    service_uuid_ = 0;
    method_id_ = 0;
    call_id_ = 0;
    error_id_ = (uint) RpcError.rpc_ok;
    dead_line_ = 0;
    action_ = null;
    debug_loop_ = false;
  }
  public long Timeout {
    get => dead_line_;
    set => dead_line_ = value + Util.GetNowMS();
  }
  public bool IsTimeout {
    get => timeout_;
    set => timeout_ = true;
  }
  public ulong ServiceUUID {
    get => service_uuid_;
    set => service_uuid_ = value;
  }
  public uint CallID {
    get => call_id_;
    set => call_id_ = value;
  }
  public uint MethodID {
    get => method_id_;
    set => method_id_ = value;
  }

  public Action Action {
    get { return action_; }
    set { action_ = value; }
  }
  public uint ErrorID {
    get { return error_id_; }
    set { error_id_ = value; }
  }
  public bool DebugLoop {
    get => debug_loop_;
    set {
      debug_loop_ = value;
      if (debug_loop_) {
        // 保存RPC调用到管理器
        GlobalProxyCallManager.Instance.Add(CallID, this);
      }
    }
  }
}

/// <summary>
/// 代理调用协程等待类
/// </summary>
/// <typeparam name="T">RPC方法返回值类型</typeparam>
internal class ProxyCallAwaiter<T> : ProxyCall, INotifyCompletion, IRecycle {
  /// <summary>
  /// 返回值
  /// </summary>
  T result_ = default(T);
  /// <summary>
  /// 调用超时
  /// </summary>
  bool timeout_ = false;
  /// <summary>
  /// 服务UUID
  /// </summary>
  ulong service_uuid_ = 0;
  /// <summary>
  /// 调用的方法序号
  /// </summary>
  uint method_id_ = 0;
  /// <summary>
  /// RPC调用ID
  /// </summary>
  uint call_id_ = 0;
  /// <summary>
  /// 本次调用对端返回的错误码
  /// </summary>
  uint error_id_ = (uint) RpcError.rpc_ok;
  /// <summary>
  /// 调用超时时间戳，毫秒
  /// </summary>
  long dead_line_ = 0;
  /// <summary>
  /// 调度器对象
  /// </summary>
  Action action_ = null;
  /// <summary>
  /// 是否是本地测试
  /// </summary>
  bool debug_loop_ = false;
  public ProxyCallAwaiter() {}
  public ProxyCallAwaiter<T> GetAwaiter() { return this; }
  public void OnCompleted(Action continuation) {
    action_ = continuation;
    GlobalProxyCallManager.Instance.Add(CallID, this);
    if (!debug_loop_) {
      // 保存到超时管理
      Singleton<Rpc>.Instance.AddWaiting(this);
    }
  }
  public void CheckExpired(long now) { timeout_ = (dead_line_ < now); }
  public bool IsCompleted => debug_loop_;
  public T GetResult() {
    var error_id = error_id_;
    bool timeout = timeout_;
    T result = result_;
    if (!debug_loop_) {
      Singleton<Rpc>.Instance.RemoveWaiting(this);
    }
    if ((RpcError) error_id_ != RpcError.rpc_ok || timeout_) {
      ObjectPool<ProxyCallAwaiter<T>>.Recycle(this);
    }
    // 检测调用结果
    switch ((RpcError) error_id) {
    case RpcError.rpc_exception:
      throw new RpcPeerErrorException();
    case RpcError.rpc_not_found:
      throw new RpcPeerMethodNotFoundException();
    case RpcError.rpc_timeout:
      throw new RpcCallTimeoutException();
    default:
      break;
    }
    if (timeout) {
      // 抛出异常
      throw new RpcCallTimeoutException();
    }
    ObjectPool<ProxyCallAwaiter<T>>.Recycle(this);
    return result;
  }
  public void SetResult(T result) { result_ = result; }

  public void OnReset() {
    result_ = default(T);
    timeout_ = false;
    service_uuid_ = 0;
    method_id_ = 0;
    call_id_ = 0;
    error_id_ = (uint) RpcError.rpc_ok;
    dead_line_ = 0;
    action_ = null;
    debug_loop_ = false;
  }

  public long Timeout {
    get => dead_line_;
    set => dead_line_ = value + Util.GetNowMS();
  }

  public bool IsTimeout {
    get => timeout_;
    set => timeout_ = true;
  }
  public bool IsDebugLoop {
    get => debug_loop_;
    set => debug_loop_ = value;
  }
  public ulong ServiceUUID {
    get => service_uuid_;
    set => service_uuid_ = value;
  }
  public uint CallID {
    get => call_id_;
    set => call_id_ = value;
  }
  public uint MethodID {
    get => method_id_;
    set => method_id_ = value;
  }
  public Action Action {
    get { return action_; }
    set { action_ = value; }
  }
  public uint ErrorID {
    get { return error_id_; }
    set { error_id_ = value; }
  }

  public bool DebugLoop {
    get => debug_loop_;
    set {
      debug_loop_ = value;
      if (debug_loop_) {
        // 保存RPC调用到管理器
        GlobalProxyCallManager.Instance.Add(CallID, this);
      }
    }
  }
}

/// <summary>
/// RPC方法参数类生成注册
/// </summary>
internal class ParamRegister {
  Dictionary<ulong, Dictionary<uint, NewParamByUUID>> register_ =
      new Dictionary<ulong, Dictionary<uint, NewParamByUUID>>();
  public Dictionary<ulong, Dictionary<uint, NewParamByUUID>> Register =>
      register_;
  public CallParam Create(ulong uuid, uint method_id) {
    Dictionary<uint, NewParamByUUID> methods;
    if (!register_.TryGetValue(uuid, out methods)) {
      return null;
    } else {
      NewParamByUUID method;
      if (!methods.TryGetValue(method_id, out method)) {
        return null;
      }
      return method();
    }
  }
}

/// <summary>
/// RPC方法返回值类生成注册
/// </summary>
internal class ReturnRegister {
  Dictionary<ulong, Dictionary<uint, NewReturnByUUID>> register_ =
      new Dictionary<ulong, Dictionary<uint, NewReturnByUUID>>();
  public Dictionary<ulong, Dictionary<uint, NewReturnByUUID>> Register =>
      register_;
  public CallReturn Create(ulong uuid, uint method_id) {
    Dictionary<uint, NewReturnByUUID> methods;
    if (!register_.TryGetValue(uuid, out methods)) {
      return null;
    } else {
      NewReturnByUUID method;
      if (!methods.TryGetValue(method_id, out method)) {
        return null;
      }
      return method();
    }
  }
}
}

namespace kratos {
/// <summary>
/// RPC工厂
/// </summary>
public static class RpcFactory {
  public static IRpc create() { return rpc.Singleton<rpc.Rpc>.Instance; }
}
}
